package m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;

/**
 * @ClassName Producer
 * @Description:类描述
 * @Author Tedu
 * @Date 2021/7/6 9:25
 * @Version V1.0
 **/
public class Producer {
    public static void main(String[] args) throws MQClientException {
        //创建事务消息生产者实例
        TransactionMQProducer t = new TransactionMQProducer("producerGroup3");


        //设置name server
        t.setNamesrvAddr("192.168.64.141:9876");

        //设置事务消息监听器
        t.setTransactionListener(new TransactionListener() {
            //执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.out.println("执行本地事务,业务参数: "+arg);
                if (true){
                    //一般是由于网络原因或者是本地事务运行缓慢
                    //rocketmq得不到事务状态,本判断仅仅测试
                    return LocalTransactionState.UNKNOW;
                }
                if(Math.random()<0.5){
                    System.out.println("事务执行成功,向RocketMQ服务器提交消息");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else {
                    System.out.println("事务执行成功,向RocketMQ服务器撤回(回滚)消息");
                    return LocalTransactionState.ROLLBACK_MESSAGE;

                }

                //执行本地事务的方法中一般不会出现未知状态,如果出现会一直回查
               // return LocalTransactionState.UNKNOW;

            }
                //处理rocketmq的事务状态回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("正在处理RocketMQ服务器的事务回查");
                return LocalTransactionState.UNKNOW;
            }
        });

        //启动
        t.start();

        //发送事务消息,会触发执行事务消息监听器来执行本地事务
        while (true){
            System.out.println("输入消息: ");
            String s = new Scanner(System.in).nextLine();
            Message msg = new Message("Topic3", s.getBytes());
            //触发监听器执行本地业务时向他传递的业务
            t.sendMessageInTransaction(msg, "假的业务数据");
        }

    }
}
